package com.jinghang.spark_base._030_SQLDataSource

import org.apache.spark.sql.SparkSession

object _010_BasicDataSourceExample {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("Spark SQL data sources example")
      .master("local[1]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()

    runBasicDataSourceExample(spark)
  }

  private def runBasicDataSourceExample(spark: SparkSession): Unit = {

    //把parquet文件转化为json文件
    val usersDF = spark.read.load("data/practiceOperator/users.parquet")
    usersDF.show()
    usersDF
      .select("name", "favorite_color")
      .write
      .json("data/out/namesAndFavColors.json")


//把json文件转化为parquet文件
    val peopleDF = spark
      .read
      .format("json")
      .load("data/practiceOperator/people.json")

    peopleDF
      .select("name", "age")
      .write.format("parquet")
      .save("data/out/namesAndAges.parquet")



    val peopleDFCsv = spark.read.format("csv")
      .option("sep", ";")//定义分隔符
      .option("inferSchema", "true")
      .option("header", "true")
      .load("data/practiceOperator/people.csv")

    peopleDFCsv.show()



    usersDF.write.format("orc")
      .option("orc.bloom.filter.columns", "favorite_color")//指定索引字段
      .option("orc.column.encoding.direct", "name")
      .save("data/out/users_with_options.orc")



    val sqlDF = spark.sql("SELECT * FROM parquet.`data/practiceOperator/users.parquet`")

    peopleDF.write.bucketBy(2, "name")
      .sortBy("age")
      .saveAsTable("people_bucketed")


    usersDF
      .write
      .partitionBy("favorite_color")
      .format("parquet")
      .save("data/out/namesPartByColor.parquet")


    usersDF
      .write
      .partitionBy("favorite_color")
      .bucketBy(2, "name")
      .saveAsTable("users_partitioned_bucketed")

    //spark.sql("DROP TABLE IF EXISTS people_bucketed")
    //spark.sql("DROP TABLE IF EXISTS users_partitioned_bucketed")
  }

}
